# This Python file uses the following encoding: utf-8
############################################################################
# Copyright (c) 2025 Li Auto Inc. and its affiliates
# Licensed under the Apache License, Version 2.0(the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
############################################################################
import os
import re
import json
from typing import Dict
from collections import deque
from openpyxl import Workbook

from trace_data import CoreData, TaskData, RunnableData, IrqData
from utils import ExcelUtils
from trace_parser import VersionStr, VersionNumber, VERSION_MAP, TRACE_DATA_PARSER

IDLE_TASK_PRIORITY = 0
EVENT_CNT_PER_TASK = 64
RUNNABLE_ID_OFFSET = 1000  # runnable id的偏移量
RUNNABLE_PRIORITY = -1  # runnable优先级


class TraceSerialCmd:
    """trace串口命令发送"""
    OPEN_ALL_EVENTS_TYPE = b"trace start\n"
    CLOSE_ALL_EVENTS_TYPE = b"trace stop\n"
    OPEN_CONNECTION_TYPE = b"trace dumpudp -s\n"
    CLOSE_CONNECTION_TYPE = b"trace dumpudp -e\n"
    SHAKE_HAND_TYPE = b"trace handshake\n"
    TRACE_FILTER_IRQ_TYPE = b"trace dumpudp irq +*\n"
    INVALIED_CORE_RESPONSE = b"error cpu id"


class TraceCmdUtils:
    """处理所有上位机和下位机的交互报文"""
    TASK_INFO_TYPE = 0x00
    OTHER_INFO_TYPE = 0x01
    RUNNABLE_INFO_TYPE = 0x02
    IRQ_INFO_TYPE = 0x03
    CMD_STATUS_TYPE = 0xe0
    OPEN_CONNECTION_TYPE = 0xff
    CLOSE_CONNECTION_TYPE = 0xfe
    OPEN_ALL_EVENTS_TYPE = 0xf1
    CLOSE_ALL_EVENTS_TYPE = 0xf0
    SNAPSHOT_CMD = 0x4  # 定义 snapshot 命令类型

    FTRACE_CMD = 0xff

    TASK_INFO_LEN = 80
    RUNNABLE_INFO_LEN = 76
    IRQ_INFO_LEN = 72
    OTHER_INFO_LEN = 8
    CMD_STATUS_LEN = 8

    @classmethod
    def get_ctrl_cmd(cls, cmd_type, param_list=[], cmd_cnt=0):
        """组装获取控制命令报文"""
        if len(param_list) == 0:
            ctrl_cmd = bytearray([4, cls.FTRACE_CMD, cmd_type, cmd_cnt])
        else:
            length = 8 + 4 * len(param_list)
            ctrl_cmd = bytearray([length, cls.FTRACE_CMD, cmd_type, cmd_cnt])
            for i in range(len(param_list)):
                ctrl_cmd += bytearray(param_list[i].to_bytes(4, byteorder="little"))
        return ctrl_cmd

    @classmethod
    def parse_ctrl_response(cls, data_list, device_data):
        """解析下位机对控制命令的回复报文"""
        res: dict = {}
        trace_data_parser = TraceInfoUtils.get_trace_parser_from_device_data(device_data)
        if trace_data_parser is None:
            return res
        trace_data_list = trace_data_parser.get_trace_data_list(data_list)
        cur_idx: int = 0
        for trace_data in trace_data_list:
            if trace_data[cur_idx + 1] != cls.FTRACE_CMD:
                continue
            response_type = trace_data[cur_idx + 2]
            if response_type == cls.OTHER_INFO_TYPE:
                res["stm_frequency"] = cls.parse_freq(trace_data)
            elif response_type == cls.TASK_INFO_TYPE:
                res.setdefault("task_info", {}).update(cls.parse_task_info(trace_data))
            elif response_type == cls.RUNNABLE_INFO_TYPE:
                runnable_data = cls.parse_runnable_info(trace_data)
                if len(runnable_data) > 0:
                    measurement_id = runnable_data["measurement_id"]
                    if measurement_id not in res.setdefault("runnable_info", {}):
                        res["runnable_info"][measurement_id] = {
                            "period_time": runnable_data["period_time"], 
                            "name": runnable_data["name"],
                            "task_id_list": [runnable_data["task_id"]]
                        }
                    else:
                        res["runnable_info"][measurement_id]["task_id_list"].append(runnable_data["task_id"])
            elif response_type == cls.IRQ_INFO_TYPE:
                res.setdefault("irq_info", {}).update(cls.parse_irq_info(trace_data))
            elif response_type == cls.CMD_STATUS_TYPE:
                cmd_cnt = cls.parse_cmd_status(trace_data)
                if cmd_cnt is not None:
                    res.setdefault("cmd_status", []).append(cmd_cnt)
            else:
                pass

        return res

    @classmethod
    def parse_cmd_status(cls, payload_list):
        """解析cmd_cnt字段"""
        if len(payload_list) != cls.CMD_STATUS_LEN:
            return None
        return payload_list[3]

    @classmethod
    def parse_freq(cls, payload_list) -> int:
        """解析下位机STM频率"""
        if len(payload_list) != cls.OTHER_INFO_LEN:
            return 0
        return int.from_bytes(payload_list[4:8], byteorder="little")

    @classmethod
    def parse_task_info(cls, payload_list) -> dict:
        """解析下位机task信息"""
        if len(payload_list) != cls.TASK_INFO_LEN:
            return {}
        task_id = int.from_bytes(payload_list[8:12], byteorder="little")
        name_data = payload_list[16:]
        name_end_idx = name_data.find(b"\0")
        if name_end_idx != -1:
            name_data = name_data[:name_end_idx]
        res = {
            task_id: 
            {
                "priority": payload_list[4], 
                "core_id": payload_list[5], 
                "name": name_data.decode("utf-8"),
            }
        }
        return res

    @classmethod
    def parse_runnable_info(cls, payload_list) -> dict:
        """解析下位机runnable信息"""
        if len(payload_list) != cls.RUNNABLE_INFO_LEN:
            return {}
        task_id = int.from_bytes(payload_list[4:8], byteorder="little")
        period_time = int.from_bytes(payload_list[72:74], byteorder="little")
        measurement_id = int.from_bytes(payload_list[74:76], byteorder="little")
        name_data = payload_list[8:72]
        name_end_idx = name_data.find(b"\0")
        if name_end_idx != -1:
            name_data = name_data[:name_end_idx]
        res = {
            "measurement_id": measurement_id, 
            "task_id": task_id, 
            "period_time": period_time, 
            "name": name_data.decode("utf-8")
        }
        return res

    @classmethod
    def parse_irq_info(cls, payload_list) -> dict:
        """解析下位机ISR信息"""
        if len(payload_list) != cls.IRQ_INFO_LEN:
            return {}
        irq_id = int.from_bytes(payload_list[5:6], byteorder="little")
        name_data = payload_list[8:]
        name_end_idx = name_data.find(b"\0")
        if name_end_idx != -1:
            name_data = name_data[:name_end_idx]
        res = {irq_id: {"core_id": payload_list[4], "name": name_data.decode("utf-8")}}
        return res


class TraceInfoUtils:
    """通用工具"""
    @classmethod
    def parse_device_conf(cls, device_conf_path):
        """解析device_conf配置信息"""
        device_config_info = {}
        if os.path.exists(device_conf_path) is False:
            print(f"[ERROR] Device config file '{device_conf_path}' does not exist, please check it.")
            return None
        with open(device_conf_path, "r", encoding="utf-8") as f:
            data = json.load(f)
        if isinstance(data, dict) is False:
            print(f"[ERROR] The content of '{device_conf_path}' should be a dict, please check it.")
            return None
        supported_version_list = [x.value for x in VersionStr]
        device_id_list = []
        supported_comm_modes = ["eth", "can"]
        for device_name, device_data in data.items():
            for _ in ["version_no", "device_id", "comm_mode"]:
                if _ not in device_data:
                    print(f"[ERROR] In '{device_conf_path}', the {device_data} does not contain '{_}' key.")
                    return None
            version_no = device_data["version_no"].lower()
            if version_no not in supported_version_list:
                print(f"[ERROR] In '{device_conf_path}', the 'version_no' must be one of {supported_version_list}.")
                return None
            device_id = device_data["device_id"]
            if device_id in device_id_list:
                print(f"[ERROR] In '{device_conf_path}', the 'device_id' {device_id} appears multiple times.")
                return None
            device_id_list.append(device_id)
            comm_mode = device_data["comm_mode"].lower()
            if comm_mode not in supported_comm_modes:
                print(f"[ERROR] In '{device_conf_path}', the 'comm_mode' must be one of {supported_comm_modes}.")
                return None
            if comm_mode == "eth":
                for _ in ["ip", "port"]:
                    if _ not in device_data:
                        print(f"[ERROR] In '{device_conf_path}', the {device_data} does not contain '{_}' key.")
                        return None
                device_data["port"] = int(device_data["port"])
            elif comm_mode.lower() == "can":
                for _ in ["can_channel", "tx_id", "rx_id"]:
                    if _ not in device_data:
                        print(f"[ERROR] In '{device_conf_path}', the {device_data} does not contain '{_}' key.")
                        return None
                for _ in ["tx_id", "rx_id"]:
                    if (isinstance(device_data[_], str) is True) and (device_data[_].startswith(("0x", "0X"))):
                        device_data[_] = int(device_data[_], 16)
                    else:
                        device_data[_] = int(device_data[_])
                device_data["can_channel"] = int(device_data["can_channel"])
            if version_no == VersionStr.VCOS_200.value:
                exclusive_keys = ["serial_port", "serial_baudrate", "core_num", "stm_frequency"]
                for k in exclusive_keys:
                    if k not in device_data:
                        print(f"[ERROR] In '{device_conf_path}', if you choose '{version_no}', \
then there must be keys {exclusive_keys}. Please check it.")
                        return None
                device_data["core_num"] = int(device_data["core_num"])
            device_data["version_no"] = VERSION_MAP[version_no]
            device_data["name"] = device_name
            device_config_info[device_id] = device_data
        print(f"[INFO] Parse device config file '{device_conf_path}' successfully.")
        return device_config_info

    @classmethod
    def parse_config_file(cls, config_file):
        """解析device配置文件,包括task、runnable、isr等信息"""
        print(f"[INFO] Parsing config file '{config_file}'...")
        if os.path.exists(config_file) is False:
            print(f"[ERROR] Config file {config_file} does not exist.")
            return None
        with open(config_file, "r", encoding="utf-8") as f:
            data = json.load(f)
        if isinstance(data, dict) is False:
            print(f"[ERROR] Content of config file {config_file} must be a dictionary.")
            return None
        device_info = {}
        for device_name, device_data in data.items():
            if isinstance(device_data, dict) is False:
                print(f"[ERROR] In config file {config_file}, content of '{device_name}' must be a dictionary.")
                return None
            for key in ["device_id", "stm_frequency", "Task", "Runnable", "ISR"]:
                if key not in device_data:
                    print(f"[ERROR] In config file {config_file}, {key} is not in {device_name}'s dictionary.")
                    return None
            task_dict, runnable_dict, irq_dict = {}, {}, {}
            for k, items in device_data["Task"].items():
                task_dict[int(k)] = TaskData(name=items[0], priority=items[1], core_id=items[2], \
                                             task_id=int(k), ready_state=True)
            for k, items in device_data["Runnable"].items():
                runnable_dict[int(k)] = RunnableData(name=items[0], task_id_list=items[1])
            for k, v in device_data["ISR"].items():
                irq_dict[int(k)] = IrqData(name=v)
            device_info[device_name] = {
                "device_id": device_data["device_id"],
                "stm_frequency": device_data["stm_frequency"],
                "Task": task_dict,
                "Runnable": runnable_dict,
                "ISR": irq_dict,
                "first_utc": device_data["first_utc"] if "first_utc" in device_data else 0,
                "version_no": device_data["version_no"],
            }
        return device_info

    @classmethod
    def get_trace_parser_from_device_data(cls, device_data):
        if "version_no" not in device_data:
            print("[ERROR] 'version_no' not in device_data.")
            return None
        version_no = device_data["version_no"]
        if version_no not in TRACE_DATA_PARSER:
            print(f"[ERROR] Trace data's version({version_no}) is not supported.")
            return None
        return TRACE_DATA_PARSER[version_no]

    @classmethod
    def check_device_name(cls, device_info, device_name):
        if device_name not in device_info:
            print(f"[ERROR] device_name '{device_name}' is wrong.")
            return False
        return True

    @classmethod
    def check_task_name(cls, device_info, device_name, task_name):
        """检查task是否存在于deivce中"""
        if cls.check_device_name(device_info, device_name) is False:
            return False
        for task_id, task_data in device_info[device_name]["Task"].items():
            if task_data.name == task_name:
                return True
        print(f"[ERROR] Task '{task_name}' not in device '{device_name}'.")
        return False

    @classmethod
    def check_irq_name(cls, device_info, device_name, irq_name):
        """检查irq是否存在于deivce中"""
        if cls.check_device_name(device_info, device_name) is False:
            return False
        if device_info[device_name]["version_no"] == VersionNumber.VCOS_200.value:
            return True
        for irq_id, irq_data in device_info[device_name]["ISR"].items():
            if irq_data.name == irq_name:
                return True
        print(f"[ERROR] ISR '{irq_name}' not in device '{device_name}'.")
        return False

    @classmethod
    def check_runnable_name(cls, device_info, device_name, task_name, runnable_name):
        """检查runnable是否存在于deivce的task中"""
        if cls.check_task_name(device_info, device_name, task_name) is False:
            return False
        for runnable_id, runnable_data, in device_info[device_name]["Runnable"].items():
            if (runnable_data.name == runnable_name):
                for task_id in runnable_data.task_id_list:
                    if task_id in device_info[device_name]["Task"] and \
                        device_info[device_name]["Task"][task_id].name == task_name:
                        return True
        print(f"[ERROR] Runnable '{runnable_name}' not in task '{task_name}'.")
        return False


class FtraceUtils:
    """ftrace文本处理工具"""
    @classmethod
    def extract_time_stamp_from_text(cls, line: str):
        if "#" in line:
            return None
        try:
            start = line.find("]") + 1
            end = line.find(":", start)
            time_stamp = float(line[start:end].strip())
            return time_stamp, line
        except ValueError:
            return None

    @classmethod
    def extract_core_id_from_text(cls, line: str):
        if "#" in line:
            return None
        try:
            start = line.find("[")
            end = line.find("]", start)
            core_id = int(line[start + 1:end].strip())
            return core_id
        except ValueError:
            return None

    @classmethod
    def ftrace_get_sorted_text(cls, input_file: str):
        """读取一个ftrace文件中的文本,且按时间戳递增排序"""
        if os.path.exists(input_file) is False:
            return []
        with open(input_file, "r", encoding="utf-8") as file:
            lines = file.readlines()
        lines_to_sort = []
        lines_to_keep = []
        for line in lines:
            key = cls.extract_time_stamp_from_text(line)
            if key:
                lines_to_sort.append(key)
            else:
                lines_to_keep.append(line)
        sorted_lines = sorted(lines_to_sort)
        final_lines = lines_to_keep + [line for time_stamp, line in sorted_lines]
        return final_lines

    @classmethod
    def ftrace_text_sort(cls, input_dir: str) -> bool:
        """对ftrace文本的时间戳做递增排序"""
        if os.path.isdir(input_dir) is False:
            print(f"[ERROR] input_dir {input_dir} does not exist or it's not a directory.")
            return False
        input_path_list = [
            os.path.join(input_dir, filename)
            for filename in os.listdir(input_dir)
            if filename.startswith("ftrace_") and (filename.endswith(".log") or filename.endswith(".txt"))
        ]
        print("[INFO] Start to sort ftrace files...")
        for input_path in input_path_list:
            print(f"[INFO] Sort {input_path}")
            final_lines = cls.ftrace_get_sorted_text(input_path)
            with open(input_path, "w") as f:
                f.writelines(final_lines)
        print("[INFO] Sort done.")
        return True

    @classmethod
    def ftrace_text_merge(cls, input_dir: str) -> bool:
        """对指定目录内的所有符合格式要求的ftrace文件做合并处理"""
        if os.path.isdir(input_dir) is False:
            print(f"[ERROR] input_dir {input_dir} does not exist or it's not a directory.")
            return False
        input_path_list = [
            os.path.join(input_dir, filename)
            for filename in os.listdir(input_dir)
            if filename.startswith("ftrace_") and
            (filename.endswith(".log") or filename.endswith(".txt")) and "merge" not in filename
        ]
        print("[INFO] Start to merge ftrace files...")
        ftrace_header = ["# tracer: nop\n", "#\n"]
        output_path = os.path.join(input_dir, "ftrace_merge.log")
        with open(output_path, "w") as f:
            f.writelines(ftrace_header)

        for input_path in input_path_list:
            print(f"[INFO] merge {input_path}")
            with open(input_path, "r", encoding='utf-8') as file:
                lines = file.readlines()
            lines_to_merge = []
            for line in lines:
                if line in ftrace_header:
                    continue
                lines_to_merge.append(line)

            with open(output_path, "a") as f:
                f.writelines(lines_to_merge)
        print("[INFO] Merge done.")
        return True

    @classmethod
    def ftrace_text_add_prefix(cls, input_dir: str) -> bool:
        """对ftrace文本的runnable名称增加prefix前缀"""
        if os.path.isdir(input_dir) is False:
            print(f"[ERROR] input_dir {input_dir} does not exist or it's not a directory.")
            return False
        input_path_list = [
            os.path.join(input_dir, filename)
            for filename in os.listdir(input_dir)
            if filename.startswith("ftrace_") and (filename.endswith(".log") or filename.endswith(".txt"))
        ]
        print("[INFO] Start to add prefix for ftrace files...")
        for input_path in input_path_list:
            print(f"[INFO] add prefix for {input_path}")
            with open(input_path, "r", encoding='utf-8') as file:
                lines = file.readlines()

            lines_to_add_prefix = []
            for line in lines:
                core_id = cls.extract_core_id_from_text(line)
                matches = re.findall(r'runnable_\d+', line)
                if len(matches) != 0 and core_id is not None:
                    runnable_name = matches[0]
                    start = line.find(runnable_name, 0)
                    # 防止多次调用产生重复的前缀
                    if start == 0 or line[start - 1] != '_':
                        line = line.replace(runnable_name, f"Core{core_id}_{runnable_name}")

                lines_to_add_prefix.append(line)

            with open(input_path, "w") as f:
                f.writelines(lines_to_add_prefix)
        print("[INFO] Add prefix done.")
        return True


class TraceComputeUtils:
    """计算trace数据的指标"""
    @classmethod
    def write_result(cls, file_path: str, core_dict, task_dict, runnable_dict, event_dict):
        """保存计算结果"""
        with open(file_path, "w") as f:
            # cpuuse
            cpuuse_dict = cls.compute_total_cpuuse(core_dict, task_dict)
            f.write(
                f"{'core_id':<10}{'cpuuse(%)':<11}{'total_time(s)':<20}{'idle_time(s)':<20}{'irq_in_idle_time(s)':<20}\n"
            )
            for core_id, cpuuse in cpuuse_dict.items():
                core_data = core_dict[core_id]
                f.write(f"{core_id:<10}{cpuuse * 100:<11.2f}{core_data.total_time * 1e-9:<20}\
{core_data.idle_time * 1e-9:<20}{core_data.irq_in_idle_time * 1e-9:<20}\n")
            f.write("\n---------------------------------------------------------------\n")
            f.flush()
            for core_id, cpuuse in cpuuse_dict.items():
                core_data = core_dict[core_id]
                f.write(f"Core {core_id}:\n")
                f.write(f"cpuuse: {cpuuse * 100:.2f}, total_time(s): {core_data.total_time * 1e-9}, \
idle_time(s): {core_data.idle_time * 1e-9}, irq_in_idle_time(s): {core_data.irq_in_idle_time * 1e-9}\n")
                # task
                f.write(f"{'TaskName':<34}{'TaskID':<7}{'Prio':<6}{'PeriodAvg(ms)':<14}\
ExcTimeMax(us) ExcTimeMin(us) ExcTimeAvg(us) ExcTimeTotal(ms)\n")
                core_data.task_id_list.sort()
                for task_id in core_data.task_id_list:
                    task_data = task_dict[task_id]
                    f.write(f"{task_data.name:<34}{task_id:<7}{task_data.priority:<6}\
{(task_data.period_total / task_data.period_count) * 1e-6 if task_data.period_count > 0 else 0:<14.6f}\
{int(task_data.execute_time_max * 1e-3):<15}{int(task_data.execute_time_min * 1e-3):<15}\
{int((task_data.execute_time_total / task_data.execute_count) * 1e-3) if task_data.execute_count > 0 else 0:<15}\
{task_data.execute_time_total * 1e-6:<16.6f}\n"
                           )
                # runnable
                f.write(f"\n{'RunnableName':<36}{'Prio':<5} {'MID':<6} Start(ms)   \
Period(ms) PeriodAvg(ms) JitterMax(us) JitterAvg(us)\n")
                for task_id in core_data.task_id_list:
                    task_data = task_dict[task_id]
                    f.write(f"{task_data.name:<36}{task_data.priority:<5} \n")
                    for mid, runnable_data in runnable_dict.items():
                        if runnable_data.task_id != task_id:
                            continue
                        f.write(f"  {runnable_data.name:<34}      {mid:<6} \
{runnable_data.start_time * 1e-6:<9.3f}              \
{(runnable_data.period_total / runnable_data.period_count) * 1e-6 if runnable_data.period_count > 0 else 0:<13.6f}\n")
                # event
                f.write(f"\n{'EventName':<36}{'Prio':<5} EvenId Start(ms) Period(ms) PeriodAvg(ms)\n")
                for task_id in core_data.task_id_list:
                    task_data = task_dict[task_id]
                    f.write(f"{task_data.name:<36}{task_data.priority}\n")
                    if task_id not in event_dict:
                        continue
                    for event_id, event_data in event_dict[task_id].items():
                        f.write(f"  {event_data.name:<34}      {event_id:<6} \
{event_data.start_time * 1e-6:<9.3f}              \
{(event_data.period_total / event_data.period_count) * 1e-6 if event_data.period_count > 0 else 0:<13.6f}\n")
                f.write("\n---------------------------------------------------------------\n")
                f.flush()

    @classmethod
    def compute_total_cpuuse(cls, core_dict: Dict[int, CoreData], task_dict: Dict[int, TaskData]):
        """总体cpuuse和时间"""
        cpuuse_dict = {}
        for core_id, core_data in core_dict.items():
            if core_data.idle_task_id != -1:
                core_data.idle_time = task_dict[core_data.idle_task_id].execute_time_total
            if core_data.total_time != 0:
                cpuuse_dict[core_id] = 1 - (core_data.idle_time - core_data.irq_in_idle_time) / core_data.total_time
            else:
                print(f"[WARNING] compute_total_cpuuse(): core{core_id}'s total_time is 0.")
                cpuuse_dict[core_id] = 0
        return {key: cpuuse_dict[key] for key in sorted(cpuuse_dict)}

    @classmethod
    def compute_task_cpuuse(cls, task_id: int, task_dict: Dict[int, TaskData], core_dict: Dict[int, CoreData]):
        """单个task cpuuse"""
        task_data = task_dict[task_id]
        for core_id, core_data in core_dict.items():
            if task_id in core_data.task_id_list:
                if core_data.total_time != 0:
                    return (task_data.execute_time_total - task_data.irq_time) / core_data.total_time
                else:
                    print(f"[WARNING] compute_task_cpuuse(): core{core_id}'s total_time is 0.")
                    return 0
        return -1

    @classmethod
    def compute_all_task_cpuuse(cls, task_dict: Dict[int, TaskData], core_dict: Dict[int, CoreData]):
        """所有task cpuuse"""
        task_cpuuse_dict = {}
        for core_id, core_data in core_dict.items():
            for task_id in core_data.task_id_list:
                task_cpuuse_dict[task_id] = cls.compute_task_cpuuse(task_id, task_dict, core_dict)
        return task_cpuuse_dict

    @classmethod
    def check_input_parameter(cls, data_dir: str, input_dict: Dict[str, list]):
        """检查API入参是否正确"""
        if os.path.isdir(data_dir) is False:
            print(f"[ERROR] data_dir '{data_dir}' does not exist.")
            return False
        for _, threshold_list in input_dict.items():
            if isinstance(threshold_list, list) is False or len(threshold_list) != 2:
                print("Parameter input_dict is illegal.")
                return False
        return True

    @classmethod
    def get_trace_data_by_device(cls, data_dir):
        """按device获取二进制trace数据"""
        # 解析config.json文件
        config_file = os.path.join(data_dir, "config.json")
        conf_dict = TraceInfoUtils.parse_config_file(config_file)
        if conf_dict is None:
            return False, None
        # 获取二进制trace数据
        trace_data_dict = {}
        for device_name, device_data in conf_dict.items():
            bin_file = os.path.join(data_dir, f"{device_name}_data.bin")
            if os.path.exists(bin_file) is False:
                print(f"[ERROR] '{bin_file}' does not exist.")
                return False, None
            trace_data_parser = TraceInfoUtils.get_trace_parser_from_device_data(device_data)
            if trace_data_parser is None:
                return False, None
            with open(bin_file, "rb") as f:
                bin_data = f.read()
            bin_data = bytearray(bin_data)
            trace_data_list = trace_data_parser.get_trace_data_list(bin_data)
            # 二进制报文按时间戳递增排序
            trace_data_list = trace_data_parser.sort_trace_data_list(trace_data_list)
            trace_data_dict[device_name] = (device_data, bin_file, trace_data_list)
        return True, trace_data_dict

    @classmethod
    def trace_get_task_statistics(cls, data_dir: str, input_task_dict: Dict[str, list]):
        """根据bin文件计算task的统计信息"""
        # 检查入参
        if cls.check_input_parameter(data_dir, input_task_dict) is False:
            return False, None
        # 按device获取二进制trace数据
        res, trace_data_dict = cls.get_trace_data_by_device(data_dir)
        if res is False:
            return False, None
        # 计算指标
        ret_task_dict = {}
        for device_name, items in trace_data_dict.items():
            device_data, bin_file, trace_data_list = items
            trace_data_parser = TraceInfoUtils.get_trace_parser_from_device_data(device_data)
            if trace_data_parser is None:
                return False, None
            trace_data_parser.tick_to_ns = 1e9 / device_data["stm_frequency"]
            print(f"[INFO] Parsing trace data file '{bin_file}'...")
            res, ret_task_dict[device_name] = \
                cls.compute_task_statistics(input_task_dict, device_data, trace_data_list, trace_data_parser)
            if res is False:
                return False, None
        return True, ret_task_dict

    @classmethod
    def compute_task_statistics(cls, input_task_dict, device_data, trace_data_list, trace_data_parser):
        """计算单个设备的task统计信息"""
        task_info_dict: Dict[str, TaskData] = {}
        task_cet_list_dict: Dict[str, list] = {}
        task_cet_dict: Dict[str, list] = {}
        task_dt_dict: Dict[str, list] = {}
        task_dt_list_dict: Dict[str, list] = {}
        reactivate_task_state: Dict[int, bool] = {} # task是否被reactivate
        irq_info_dict: Dict[int, IrqData] = {}
        irq_stack_dict = {}

        for trace_data in trace_data_list:
            event_len: int = trace_data[0]
            if event_len > len(trace_data):
                print(f"[ERROR] compute_task_statistics(): length of the event({event_len}) is wrong.")
                return False, None
            event_type = trace_data[1]
            if event_type == trace_data_parser.FTRACE_SCHED_SWITCH:
                result = trace_data_parser.parse_sched_switch(event_len, trace_data)
                if not result:
                    return False, None
                _, _, next_task_id, time_stamp, prev_task_id, prev_state = result
                # 切出task
                if prev_task_id not in device_data["Task"]:
                    print(f"[ERROR] {prev_task_id} not in 'Task' dict of config file.")
                    return False, None
                prev_task_name = device_data["Task"][prev_task_id].name
                if prev_task_name not in task_info_dict:
                    task_info_dict[prev_task_name] = TaskData()
                prev_task_data = task_info_dict[prev_task_name]
                prev_task_data.last_end_time = time_stamp
                # 累加切出task本次的CET时间
                if prev_task_data.last_execute_time > 0:
                    prev_task_data.execute_time_cal += prev_task_data.last_end_time - prev_task_data.last_execute_time
                # 切出task是否已结束执行
                if (prev_task_id in reactivate_task_state) and (reactivate_task_state[prev_task_id] is True):
                    prev_state = trace_data_parser.TASK_STATE_SUSPEND
                    reactivate_task_state[prev_task_id] = False
                if trace_data_parser.is_task_blocked(prev_state) is True:
                    if prev_task_data.last_start_time > 0:
                        # CET
                        prev_task_data.execute_time_total += prev_task_data.execute_time_cal
                        prev_task_data.execute_time_max = \
                            max(prev_task_data.execute_time_max, prev_task_data.execute_time_cal)
                        prev_task_data.execute_time_min = \
                            min(prev_task_data.execute_time_min, prev_task_data.execute_time_cal) \
                            if prev_task_data.execute_time_min != 0 else prev_task_data.execute_time_cal
                        prev_task_data.execute_count += 1
                        task_cet_list_dict.setdefault(prev_task_name, []).append(prev_task_data.execute_time_cal)
                        # CET超时时间戳
                        if (prev_task_name in input_task_dict) and (input_task_dict[prev_task_name][0] is not None) \
                            and (prev_task_data.execute_time_cal > input_task_dict[prev_task_name][0]):
                            task_cet_dict.setdefault(prev_task_name, []).append(prev_task_data.last_start_time)
                        # GET
                        gross_execution_time = prev_task_data.last_end_time - prev_task_data.last_start_time
                        prev_task_data.get_total += gross_execution_time
                        prev_task_data.get_max = max(prev_task_data.get_max, gross_execution_time)
                        prev_task_data.get_min = min(prev_task_data.get_min, gross_execution_time) if \
                            prev_task_data.get_min != 0 else gross_execution_time
                    prev_task_data.execute_time_cal = 0
                    # NOTE 实际上切出task结束执行后没有立刻ready，只是方便在无wake up事件时，
                    # 记录task ready后第一次running的时间
                    prev_task_data.ready_state = True

                # 切入task
                if next_task_id not in device_data["Task"]:
                    print(f"[ERROR] {next_task_id} not in 'Task' dict of config file.")
                    return False, None
                next_task_name = device_data["Task"][next_task_id].name
                if next_task_name not in task_info_dict:
                    task_info_dict[next_task_name] = TaskData()
                next_task_data = task_info_dict[next_task_name]
                next_task_data.last_execute_time = time_stamp
                # task结束执行后首次进running，等价于ready后首次进running
                if next_task_data.ready_state is True:
                    # DT
                    if next_task_data.last_start_time != 0:
                        period = time_stamp - next_task_data.last_start_time
                        next_task_data.period_total += period
                        next_task_data.period_max = max(next_task_data.period_max, period)
                        next_task_data.period_min = min(next_task_data.period_min, period) \
                            if next_task_data.period_min != 0 else period
                        next_task_data.period_count += 1
                        task_dt_list_dict.setdefault(next_task_name, []).append(period)
                        # DT超时时间戳
                        if (next_task_name in input_task_dict) and (input_task_dict[next_task_name][1] is not None) \
                            and (period > input_task_dict[next_task_name][1]):
                            task_dt_dict.setdefault(next_task_name, []).append(time_stamp)
                    # 记CET起点
                    next_task_data.last_start_time = time_stamp
                    next_task_data.ready_state = False
            elif event_type == trace_data_parser.FTRACE_TASK_SUSPEND:
                result = trace_data_parser.parse_suspend(event_len, trace_data)
                if not result:
                    return False, None
                core_id, prev_task_id, time_stamp, prev_state = result
                # 切出task
                if prev_task_id not in device_data["Task"]:
                    print(f"[ERROR] {prev_task_id} not in 'Task' dict of config file.")
                    return False, None
                prev_task_name = device_data["Task"][prev_task_id].name
                if prev_task_name not in task_info_dict:
                    task_info_dict[prev_task_name] = TaskData()
                prev_task_data = task_info_dict[prev_task_name]
                prev_task_data.last_end_time = time_stamp
                # 累加切出task本次的CET时间
                if prev_task_data.last_execute_time > 0:
                    prev_task_data.execute_time_cal += prev_task_data.last_end_time - prev_task_data.last_execute_time
                # 切出task是否已结束执行
                if (prev_task_id in reactivate_task_state) and (reactivate_task_state[prev_task_id] is True):
                    prev_state = trace_data_parser.TASK_STATE_SUSPEND
                    reactivate_task_state[prev_task_id] = False
                if trace_data_parser.is_task_blocked(prev_state) is True:
                    if prev_task_data.last_start_time > 0:
                        # 在ISR中suspend
                        if len(irq_stack_dict.setdefault(core_id, deque())) > 0:
                            irq_id = irq_stack_dict[core_id][0]
                            irq_data = irq_info_dict[irq_id]
                            prev_task_data.execute_time_cal -= \
                                (time_stamp - max(irq_data.last_execute_time, prev_task_data.last_start_time))
                        # CET
                        prev_task_data.execute_time_total += prev_task_data.execute_time_cal
                        prev_task_data.execute_time_max = \
                            max(prev_task_data.execute_time_max, prev_task_data.execute_time_cal)
                        prev_task_data.execute_time_min = \
                            min(prev_task_data.execute_time_min, prev_task_data.execute_time_cal) if \
                            prev_task_data.execute_time_min != 0 else prev_task_data.execute_time_cal
                        prev_task_data.execute_count += 1
                        task_cet_list_dict.setdefault(prev_task_name, []).append(prev_task_data.execute_time_cal)
                        # CET超时时间戳
                        if (prev_task_name in input_task_dict) and (input_task_dict[prev_task_name][0] is not None) \
                            and (prev_task_data.execute_time_cal > input_task_dict[prev_task_name][0]):
                            task_cet_dict.setdefault(prev_task_name, []).append(prev_task_data.last_start_time)
                        # GET
                        gross_execution_time = prev_task_data.last_end_time - prev_task_data.last_start_time
                        prev_task_data.get_total += gross_execution_time
                        prev_task_data.get_max = max(prev_task_data.get_max, gross_execution_time)
                        prev_task_data.get_min = min(prev_task_data.get_min, gross_execution_time) if \
                            prev_task_data.get_min != 0 else gross_execution_time
                    prev_task_data.execute_time_cal = 0
                    # NOTE 实际上切出task结束执行后没有立刻ready，只是方便在无wake up事件时，
                    # 记录task ready后第一次running的时间
                    prev_task_data.ready_state = True
            elif event_type == trace_data_parser.FTRACE_TASK_RESUME:
                result = trace_data_parser.parse_resume(event_len, trace_data)
                if not result:
                    return False, None
                _, core_id, next_task_id, time_stamp = result
                # 切入task
                if next_task_id not in device_data["Task"]:
                    print(f"[ERROR] {next_task_id} not in 'Task' dict of config file.")
                    return False, None
                next_task_name = device_data["Task"][next_task_id].name
                if next_task_name not in task_info_dict:
                    task_info_dict[next_task_name] = TaskData()
                next_task_data = task_info_dict[next_task_name]
                next_task_data.last_execute_time = time_stamp
                # task结束执行后首次进running，等价于ready后首次进running
                if next_task_data.ready_state is True:
                    # DT
                    if next_task_data.last_start_time != 0:
                        period = time_stamp - next_task_data.last_start_time
                        next_task_data.period_total += period
                        next_task_data.period_max = max(next_task_data.period_max, period)
                        next_task_data.period_min = min(next_task_data.period_min, period) \
                            if next_task_data.period_min != 0 else period
                        next_task_data.period_count += 1
                        task_dt_list_dict.setdefault(next_task_name, []).append(period)
                        # DT超时时间戳
                        if (next_task_name in input_task_dict) and (input_task_dict[next_task_name][1] is not None) \
                            and (period > input_task_dict[next_task_name][1]):
                            task_dt_dict.setdefault(next_task_name, []).append(time_stamp)
                    # 记CET起点
                    next_task_data.last_start_time = time_stamp
                    next_task_data.ready_state = False
            elif event_type == trace_data_parser.FTRACE_TASK_REACTIVATE:
                result = trace_data_parser.parse_reactivate(event_len, trace_data)
                if not result:
                    return False, None
                _, _, pid, time_stamp = result
                if pid not in device_data["Task"]:
                    print(f"[ERROR] {pid} not in 'Task' dict of config file.")
                    return False, None
                reactivate_task_state[pid] = True
                task_name = device_data["Task"][pid].name
                if task_name not in task_info_dict:
                    task_info_dict[task_name] = TaskData()
                task_data = task_info_dict[task_name]
                if task_data.last_ready_time != 0:
                    task_data.ready_period_total += time_stamp - task_data.last_ready_time
                    task_data.ready_cnt += 1
                task_data.last_ready_time = time_stamp
            elif event_type == trace_data_parser.FTRACE_IRQ_ENTRY:
                result = trace_data_parser.parse_irq_entry(event_len, trace_data)
                if not result:
                    return False, None
                core_id, pid, time_stamp, irq_id, ori_irq_id, _ = result
                if pid not in device_data["Task"]:
                    print(f"[ERROR] {pid} not in 'Task' dict of config file.")
                    return False, None
                if trace_data_parser.VERSION_NO == VersionNumber.VCOS_200.value:
                    device_data["ISR"][irq_id] = IrqData(name=f"ISR_{ori_irq_id}_core{core_id}")
                if irq_id not in device_data["ISR"]:
                    print(f"[ERROR] {irq_id} not in 'ISR' dict of config file.")
                    return False, None
                if irq_id not in irq_info_dict:
                    irq_info_dict[irq_id] = IrqData()
                irq_info_dict[irq_id].last_execute_time = time_stamp
                # ISR入嵌套栈
                irq_stack_dict.setdefault(core_id, deque()).append(irq_id)
            elif event_type == trace_data_parser.FTRACE_IRQ_EXIT:
                result = trace_data_parser.parse_irq_exit(event_len, trace_data)
                if not result:
                    return False, None
                core_id, pid, time_stamp, irq_id, ori_irq_id = result
                if pid not in device_data["Task"]:
                    print(f"[ERROR] {pid} not in 'Task' dict of config file.")
                    return False, None
                task_name = device_data["Task"][pid].name
                if trace_data_parser.VERSION_NO == VersionNumber.VCOS_200.value:
                    device_data["ISR"][irq_id] = IrqData(name=f"ISR_{ori_irq_id}_core{core_id}")
                if irq_id not in device_data["ISR"]:
                    print(f"[ERROR] {irq_id} not in 'ISR' dict of config file.")
                    return False, None
                if irq_id not in irq_info_dict:
                    continue
                irq_data = irq_info_dict[irq_id]
                # ISR出嵌套栈
                if core_id not in irq_stack_dict:
                    irq_stack_dict[core_id] = deque()
                irq_stack = irq_stack_dict[core_id]
                if (len(irq_stack) > 0) and (irq_stack[-1] == irq_id):
                    irq_stack.pop()
                    if (len(irq_stack) == 0) and (irq_data.last_execute_time > 0) and (task_name in task_info_dict):
                        task_data = task_info_dict[task_name]
                        task_data.execute_time_cal -= \
                            time_stamp - max(irq_data.last_execute_time, task_data.last_start_time)
            elif event_type == trace_data_parser.FTRACE_WAKE_UP:
                result = trace_data_parser.parse_sched_wakeup(event_len, trace_data)
                if not result:
                    return False, None
                _, core_id, _, time_stamp, task_id = result
                if task_id not in device_data["Task"]:
                    print(f"[ERROR] {task_id} not in 'Task' dict of config file.")
                    return False, None
                task_name = device_data["Task"][task_id].name
                if task_name not in task_info_dict:
                    task_info_dict[task_name] = TaskData()
                task_data = task_info_dict[task_name]
                if task_data.last_ready_time != 0:
                    task_data.ready_period_total += time_stamp - task_data.last_ready_time
                    task_data.ready_cnt += 1
                task_data.last_ready_time = time_stamp

        ret_task_dict = {}
        if len(input_task_dict) == 0:
            task_name_list = list(task_info_dict.keys())
        else:
            task_name_list = list(input_task_dict.keys())
        for task_name in task_name_list:
            ret_task_dict[task_name] = {
                "get_avg": 0,
                "get_max": 0,
                "get_min": 0,
                "cet_avg": 0,
                "cet_max": 0,
                "cet_min": 0,
                "cet_list": [],
                "cet_timestamp": [],  # CET超出阈值的时间点
                "dt_avg": 0,
                "dt_max": 0,
                "dt_min": 0,
                "dt_timestamp": [],  # DT超出阈值的时间点
                "period_avg": 0,
                "jitter_avg": 0,
                "jitter_max": 0,
                "jitter_min": 0,
            }
        for task_name, task_data in task_info_dict.items():
            if task_name not in ret_task_dict:
                continue
            ret_task_data = ret_task_dict[task_name]
            if task_data.execute_count > 0:
                ret_task_data["get_avg"] = task_data.get_total / task_data.execute_count
                ret_task_data["cet_avg"] = task_data.execute_time_total / task_data.execute_count
            ret_task_data["get_max"] = task_data.get_max
            ret_task_data["get_min"] = task_data.get_min
            ret_task_data["cet_max"] = task_data.execute_time_max
            ret_task_data["cet_min"] = task_data.execute_time_min
            if task_name in task_cet_list_dict:
                ret_task_data["cet_list"] = task_cet_list_dict[task_name]
            if task_name in task_cet_dict:
                ret_task_data["cet_timestamp"] = task_cet_dict[task_name]
            if task_data.period_count > 0:
                ret_task_data["dt_avg"] = task_data.period_total / task_data.period_count
            ret_task_data["dt_max"] = task_data.period_max
            ret_task_data["dt_min"] = task_data.period_min
            if task_name in task_dt_dict:
                ret_task_data["dt_timestamp"] = task_dt_dict[task_name]
            if (task_data.ready_cnt > 0) and (task_name in task_dt_list_dict):
                ready_period_avg = task_data.ready_period_total / task_data.ready_cnt
                ret_task_data["period_avg"] = ready_period_avg
                ret_task_data["jitter_avg"] = abs(ret_task_data["dt_avg"] - ready_period_avg)
                ret_task_data["jitter_max"] = max(abs(ret_task_data["dt_max"] - ready_period_avg), 
                                                abs(ret_task_data["dt_min"] - ready_period_avg))
                ret_task_data["jitter_min"] = min(abs(x - ready_period_avg) for x in task_dt_list_dict[task_name])
        return True, ret_task_dict

    @classmethod
    def trace_get_irq_statistics(cls, data_dir: str, input_irq_dict: Dict[str, list]={}):
        """根据bin文件计算irq的统计信息"""
        # 检查入参
        if cls.check_input_parameter(data_dir, input_irq_dict) is False:
            return False, None
        # 按device获取二进制trace数据
        res, trace_data_dict = cls.get_trace_data_by_device(data_dir)
        if res is False:
            return False, None
        # 计算指标
        ret_irq_dict = {}
        for device_name, items in trace_data_dict.items():
            device_data, bin_file, trace_data_list = items
            trace_data_parser = TraceInfoUtils.get_trace_parser_from_device_data(device_data)
            if trace_data_parser is None:
                return False, None
            trace_data_parser.tick_to_ns = 1e9 / device_data["stm_frequency"]
            print(f"[INFO] Parsing trace data file '{bin_file}'...")
            res, ret_irq_dict[device_name] = cls.compute_irq_statistics(
                input_irq_dict, device_data, trace_data_list, trace_data_parser
            )
            if res is False:
                return False, None
        return True, ret_irq_dict

    @classmethod
    def compute_irq_statistics(cls, input_irq_dict, device_data, trace_data_list, trace_data_parser):
        """计算单个设备的irq统计信息"""
        irq_info_dict: Dict[str, IrqData] = {}
        irq_cet_list_dict: Dict[str, list] = {}
        irq_cet_dict: Dict[str, list] = {}
        irq_dt_dict: Dict[str, list] = {}
        irq_stack_dict = {}

        for trace_data in trace_data_list:
            event_len: int = trace_data[0]
            if event_len > len(trace_data):
                print(f"[ERROR] compute_irq_statistics(): length of the event({event_len}) is wrong.")
                return False, None
            event_type = trace_data[1]
            if event_type == trace_data_parser.FTRACE_IRQ_ENTRY:
                result = trace_data_parser.parse_irq_entry(event_len, trace_data)
                if not result:
                    return False, None
                core_id, _, time_stamp, irq_id, ori_irq_id, _ = result
                # ISR嵌套栈
                if core_id not in irq_stack_dict:
                    irq_stack_dict[core_id] = deque()
                irq_stack = irq_stack_dict[core_id]
                if trace_data_parser.VERSION_NO == VersionNumber.VCOS_200.value:
                    device_data["ISR"][irq_id] = IrqData(name=f"ISR_{ori_irq_id}_core{core_id}")
                if irq_id not in device_data["ISR"]:
                    print(f"[ERROR] {irq_id} not in 'ISR' dict of config file.")
                    return False, None
                irq_name = device_data["ISR"][irq_id].name
                if irq_name not in irq_info_dict:
                    irq_info_dict[irq_name] = IrqData()
                irq_data = irq_info_dict[irq_name]
                # DT
                if irq_data.last_execute_time != 0:
                    period = time_stamp - irq_data.last_execute_time
                    irq_data.period_max = max(period, irq_data.period_max)
                    irq_data.period_min = min(period, irq_data.period_min) if irq_data.period_min != 0 else period
                    irq_data.period_total += period
                    irq_data.period_count += 1
                    # DT超时时间戳
                    if (irq_name in input_irq_dict) and (input_irq_dict[irq_name][1] is not None) and \
                        (period > input_irq_dict[irq_name][1]):
                        irq_dt_dict.setdefault(irq_name, []).append(time_stamp)
                # ISR入栈
                irq_stack.append(irq_name)
                irq_data.last_execute_time = time_stamp
            elif event_type == trace_data_parser.FTRACE_IRQ_EXIT:
                result = trace_data_parser.parse_irq_exit(event_len, trace_data)
                if not result:
                    return False, None
                core_id, _, time_stamp, irq_id, ori_irq_id = result
                # ISR嵌套栈
                if core_id not in irq_stack_dict:
                    irq_stack_dict[core_id] = deque()
                irq_stack = irq_stack_dict[core_id]
                if trace_data_parser.VERSION_NO == VersionNumber.VCOS_200.value:
                    device_data["ISR"][irq_id] = IrqData(name=f"ISR_{ori_irq_id}_core{core_id}")
                if irq_id not in device_data["ISR"]:
                    print(f"[ERROR] {irq_id} not in 'ISR' dict of config file.")
                    return False, None
                irq_name = device_data["ISR"][irq_id].name
                if irq_name not in irq_info_dict:
                    continue
                irq_data = irq_info_dict[irq_name]
                # ISR出栈
                if (len(irq_stack) > 0) and (irq_stack[-1] == irq_name):
                    irq_stack.pop()
                    if len(irq_stack) > 0:
                        top_irq_name = irq_stack[-1]
                        irq_info_dict[top_irq_name].preemption_time += time_stamp - irq_data.last_execute_time
                # CET
                execute_time = time_stamp - irq_data.last_execute_time - irq_data.preemption_time
                irq_cet_list_dict.setdefault(irq_name, []).append(execute_time)
                irq_data.execute_time_max = max(execute_time, irq_data.execute_time_max)
                irq_data.execute_time_min = min(execute_time, irq_data.execute_time_min) \
                    if irq_data.execute_time_min != 0 else execute_time
                irq_data.execute_time_total += execute_time
                irq_data.execute_count += 1
                irq_data.preemption_time = 0
                # CET超时时间戳
                if (irq_name in input_irq_dict) and (input_irq_dict[irq_name][0] is not None) and \
                    (execute_time > input_irq_dict[irq_name][0]):
                    irq_cet_dict.setdefault(irq_name, []).append(irq_data.last_execute_time)
                # GET
                gross_execution_time = time_stamp - irq_data.last_execute_time
                irq_data.get_max = max(gross_execution_time, irq_data.get_max)
                irq_data.get_min = min(gross_execution_time, irq_data.get_min) \
                    if irq_data.get_min != 0 else gross_execution_time
                irq_data.get_total += gross_execution_time

        ret_irq_dict = {}
        if len(input_irq_dict) == 0:
            irq_name_list = list(irq_info_dict.keys())
        else:
            irq_name_list = list(input_irq_dict.keys())
        for irq_name in irq_name_list:
            ret_irq_dict[irq_name] = {
                "get_avg": 0,
                "get_max": 0,
                "get_min": 0,
                "cet_avg": 0,
                "cet_max": 0,
                "cet_min": 0,
                "cet_list": [],
                "cet_timestamp": [],  # CET超出阈值的时间点
                "dt_avg": 0,
                "dt_max": 0,
                "dt_min": 0,
                "dt_timestamp": [],  # DT超出阈值的时间点
            }
        for irq_name, irq_data in irq_info_dict.items():
            if irq_name not in ret_irq_dict:
                continue
            ret_irq_data = ret_irq_dict[irq_name]
            if irq_data.execute_count > 0:
                ret_irq_data["get_avg"] = irq_data.get_total / irq_data.execute_count
                ret_irq_data["cet_avg"] = irq_data.execute_time_total / irq_data.execute_count
            ret_irq_data["get_max"] = irq_data.get_max
            ret_irq_data["get_min"] = irq_data.get_min
            ret_irq_data["cet_max"] = irq_data.execute_time_max
            ret_irq_data["cet_min"] = irq_data.execute_time_min
            if irq_name in irq_cet_list_dict:
                ret_irq_data["cet_list"] = irq_cet_list_dict[irq_name]
            if irq_name in irq_cet_dict:
                ret_irq_data["cet_timestamp"] = irq_cet_dict[irq_name]
            if irq_data.period_count > 0:
                ret_irq_data["dt_avg"] = irq_data.period_total / irq_data.period_count
            ret_irq_data["dt_max"] = irq_data.period_max
            ret_irq_data["dt_min"] = irq_data.period_min
            if irq_name in irq_dt_dict:
                ret_irq_data["dt_timestamp"] = irq_dt_dict[irq_name]
        return True, ret_irq_dict

    @classmethod
    def produce_statistics_file(cls, data_dir: str):
        """生成指标报告文件"""
        # 检查入参
        if cls.check_input_parameter(data_dir, {}) is False:
            return False
        # 按device获取二进制trace数据
        res, trace_data_dict = cls.get_trace_data_by_device(data_dir)
        if res is False:
            return False
        # 计算指标
        for device_name, items in trace_data_dict.items():
            device_data, bin_file, trace_data_list = items
            trace_data_parser = TraceInfoUtils.get_trace_parser_from_device_data(device_data)
            if trace_data_parser is None:
                return False
            trace_data_parser.tick_to_ns = 1e9 / device_data["stm_frequency"]
            task_res, task_dict = cls.compute_task_statistics({}, device_data, trace_data_list, trace_data_parser)
            irq_res, irq_dict = cls.compute_irq_statistics({}, device_data, trace_data_list, trace_data_parser)
            excel_file = os.path.join(data_dir, f"{device_name}_statistics.xlsx")
            workbook = Workbook()
            task_sheet = workbook.active
            task_sheet.title = "Task"
            if task_res is True:
                ExcelUtils.write_merged_header(task_sheet, {"name": [1, 1, 1, 1]})
                header_info = {
                    "get_avg": [1, 1, 2, 2],
                    "get_max": [1, 1, 3, 3],
                    "get_min": [1, 1, 4, 4],
                    "cet_avg": [1, 1, 5, 5],
                    "cet_max": [1, 1, 6, 6],
                    "cet_min": [1, 1, 7, 7],
                    "dt_avg": [1, 1, 8 ,8],
                    "dt_max": [1, 1, 9, 9],
                    "dt_min": [1, 1, 10, 10],
                    "period_avg": [1, 1, 11, 11],
                    "jitter_avg": [1, 1, 12, 12],
                    "jitter_max": [1, 1, 13, 13],
                    "jitter_min": [1, 1, 14, 14],
                }
                ExcelUtils.write_merged_header(task_sheet, header_info, "(ns)")
                cur_row = 1
                for task_name in sorted(task_dict.keys()):
                    items = task_dict[task_name]
                    cur_row += 1
                    ExcelUtils.write_sheet_cell(task_sheet, cur_row, 1, task_name)
                    for key in header_info.keys():
                        if key in items:
                            ExcelUtils.write_sheet_cell(task_sheet, cur_row, header_info[key][2], items[key])
                ExcelUtils.set_column_widths(task_sheet, ori_length=5)
                ExcelUtils.set_row_heights(task_sheet)
            else:
                print(f"[ERROR] Failed to parse task statistics of {device_name}.")
            irq_sheet = workbook.create_sheet(title="ISR")
            if irq_res is True:
                ExcelUtils.write_merged_header(irq_sheet, {"name": [1, 1, 1, 1]})
                header_info = {
                    "get_avg": [1, 1, 2, 2],
                    "get_max": [1, 1, 3, 3],
                    "get_min": [1, 1, 4, 4],
                    "cet_avg": [1, 1, 5, 5],
                    "cet_max": [1, 1, 6, 6],
                    "cet_min": [1, 1, 7, 7],
                    "dt_avg": [1, 1, 8 ,8],
                    "dt_max": [1, 1, 9, 9],
                    "dt_min": [1, 1, 10, 10]
                }
                ExcelUtils.write_merged_header(irq_sheet, header_info, "(ns)")
                cur_row = 1
                for irq_name in sorted(irq_dict.keys()):
                    items = irq_dict[irq_name]
                    cur_row += 1
                    ExcelUtils.write_sheet_cell(irq_sheet, cur_row, 1, irq_name)
                    for key in header_info.keys():
                        if key in items:
                            ExcelUtils.write_sheet_cell(irq_sheet, cur_row, header_info[key][2], items[key])
                ExcelUtils.set_column_widths(irq_sheet, ori_length=5)
                ExcelUtils.set_row_heights(irq_sheet)
            else:
                print(f"[ERROR] Failed to parse ISR statistics of {device_name}.")

            try:
                workbook.save(excel_file)
            except PermissionError:
                print(f"[ERROR] File {excel_file} is open in another application. Please close it and try again.")
            else:
                print(f"[INFO] Successfully to produce {excel_file}.")

